package com.talkyun.utils.sender;

import com.talkyun.utils.Looper;
import com.talkyun.utils.LooperPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class AsyncSender<T> {
    protected Logger logger = LoggerFactory.getLogger(AsyncSender.class);
    private String name;
    private int threadSize;
    private BlockingQueue<T> queue;
    private LooperPool pool;

    public AsyncSender(String senderName, int threadSize, int bufferSize) {
        this.name = senderName;
        this.threadSize = threadSize;
        this.queue = new ArrayBlockingQueue<T>(bufferSize);
    }

    public void send(T msg) {
        while (!queue.offer(msg)) {
            T oldMsg = queue.poll();
            this.handleLostData(oldMsg);
        }
    }

    protected void handleLostData(T msg) {
        // do nothing
    }

    protected void handleSendData(T msg) {
        // do nothing;
    }

    public void startup() {
        logger.info("start sender {} looper ... ", name);

        pool = new LooperPool(threadSize, new LooperPool.LooperBuilder() {
            @Override
            public Looper build() {
                return new Looper(name, 50, 1500) {
                    private T msg;

                    @Override
                    protected void loop() throws Throwable {
                        if (msg == null) {
                            msg = queue.poll(100, TimeUnit.MILLISECONDS);
                        }

                        handleSendData(msg);
                        msg = null;
                    }
                };
            }
        });

        pool.startup();
    }

    public void shutdown() {
        if (pool != null) {
            pool.shutdown();
        }
    }
}